package com.atguigu.app.dwd.db;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.func.DwdTableProcessFuction;
import com.atguigu.uitl.MyKafkaUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author hjy
 * @create 2023/3/13 18:21
 */
public class BaseDBApp {
    public static void main(String[] args) throws Exception {
        //todo 1 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        env.enableCheckpointing(5000L);
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall-flink/check");
//        env.setStateBackend(new HashMapStateBackend());
//        System.setProperty("HAOOP_USER_NAME","atguigu");
        //todo 2 从kafka(topic_db)读数据
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("topic_db", "BaseDBApp_demo"));
        //todo 3 简单过滤并转为JSONObject  主流
        SingleOutputStreamOperator<JSONObject> jsonObjectDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    JSONObject jsonObject = JSONObject.parseObject(value);
                    out.collect(jsonObject);
                }
            }
        });
        //todo 4 cdc从mysql读取配置信息
        MySqlSource<String> mysql = MySqlSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("123456")
                .databaseList("gmall_config")
                .tableList("gmall_config.table_process")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.latest())
                .build();
        DataStreamSource<String> configDS = env.fromSource(mysql, WatermarkStrategy.noWatermarks(), "mysql");
        //todo 5 将配置流转为广播流
        MapStateDescriptor<String, TableProcess> mapStateDescriptor = new MapStateDescriptor<String, TableProcess>("map-state",String.class,TableProcess.class);
        BroadcastStream<String> broadcastDS = configDS.broadcast(mapStateDescriptor);
        //todo 6 连接两个流
        BroadcastConnectedStream<JSONObject, String> connectedStream = jsonObjectDS.connect(broadcastDS);

        //todo 7 通过配置流数据过滤主流数据
        SingleOutputStreamOperator<JSONObject> processDS = connectedStream.process(new DwdTableProcessFuction(mapStateDescriptor));
        //todo 8 写出到kafka
        processDS.print("processDS>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>");
        processDS.addSink(MyKafkaUtil.getKafkaProducer());
        //todo 9 执行
        env.execute();
    }
}
